package com.navercorp.pinpoint.web.kafka;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.navercorp.pinpoint.common.hbase.HBaseTables;
import com.navercorp.pinpoint.web.dao.es.EsAgentInfoDao;
import com.navercorp.pinpoint.web.dao.es.EsApplicationIndexDao;
import com.navercorp.pinpoint.web.dao.es.entity.AgentInfoEntity;
import com.navercorp.pinpoint.web.dao.es.entity.ApplicationEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class Listener {

    @Autowired
    private EsAgentInfoDao esAgentInfoDao;

    @Autowired
    private EsApplicationIndexDao esApplicationIndexDao;

    @KafkaListener(topics = {"test"})
    public void listen(ConsumerRecord<String, String> record) throws IOException {
        log.info("kafka的key: " + record.key());
        log.info("kafka的value: " + record.value());
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        if (HBaseTables.AGENTINFO_STR.equals(record.key())) {
            AgentInfoEntity agentInfoBo = objectMapper.readValue(record.value(), AgentInfoEntity.class);
            System.out.println(agentInfoBo);
            esAgentInfoDao.insert(agentInfoBo);
            esApplicationIndexDao.insert(agentInfoBo);
        }
    }

    public static void main(String[] args) throws IOException {
        ObjectMapper objectMapper = new ObjectMapper();
        ApplicationEntity applicationEntity = new ApplicationEntity();
        applicationEntity.setServiceTypeCode((short) 10010);
        applicationEntity.setAgentId("es-guns-20190523");
        applicationEntity.setApplicationName("es-guns-0523");
        String json = objectMapper.writeValueAsString(applicationEntity);
        System.out.println(json);
        ApplicationEntity applicationEntity1 = objectMapper.readValue(json, applicationEntity.getClass());
        System.out.println(applicationEntity1);
    }
}
